package com.cctc.rds.nrpc.core.server.handler;

import com.cctc.rds.nrpc.api.rpc.RpcMethod;
import com.cctc.rds.nrpc.api.rpc.RpcRequest;
import com.cctc.rds.nrpc.api.rpc.RpcResponse;
import com.cctc.rds.nrpc.api.rpc.RpcServerRouter;
import com.cctc.rds.nrpc.core.protocol.message.impl.HeartBeatMessage;
import com.cctc.rds.nrpc.core.protocol.message.impl.RpcRequestMessage;
import com.cctc.rds.nrpc.core.protocol.message.impl.RpcResponseMessage;
import com.cctc.rds.nrpc.core.server.connnection.DefaultServerConnectionPool;
import com.cctc.rds.nrpc.core.util.PidUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.AttributeKey;

public class ServerRpcMessageHandler extends ChannelInboundHandlerAdapter {
    private RpcServerRouter router;
    private DefaultServerConnectionPool connectionPool;

    public ServerRpcMessageHandler(RpcServerRouter router, DefaultServerConnectionPool connectionPool) {
        this.router = router;
        this.connectionPool = connectionPool;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        connectionPool.put(ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        connectionPool.remove(ctx.channel());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof HeartBeatMessage) {
            HeartBeatMessage heartbeat = (HeartBeatMessage) msg;
            if (ctx.channel().attr(AttributeKey.valueOf("PID")).get() == null) {
                ctx.channel().attr(AttributeKey.valueOf("PID")).set(heartbeat.getPid());
            }
            heartbeat.setPid(PidUtils.getPid());
            ctx.writeAndFlush(heartbeat);
        } else if (msg instanceof RpcRequestMessage) {
            RpcRequestMessage rpcRequestMessage = (RpcRequestMessage) msg;
            RpcRequest rpcRequest = rpcRequestMessage.getRpcRequest();
            RpcMethod rpcMethod = router.route(rpcRequest.getInterfaces(), rpcRequest.getMethodName(), rpcRequest.getParameterTypes());
            RpcResponseMessage rpcResponseMessage = RpcResponseMessage.builder().txid(rpcRequestMessage.getTxid()).build();
            RpcResponse rpcResponse = RpcResponse.builder().build();
            try {
                Object result = rpcMethod.invoke(rpcRequest.getArguments());
                rpcResponse.setResult(result);
                rpcResponseMessage.setRpcResponse(rpcResponse);
                ctx.channel().writeAndFlush(rpcResponseMessage);
            } catch (Throwable e) {
                rpcResponse.setException(e);
                rpcResponseMessage.setRpcResponse(rpcResponse);
                ctx.channel().writeAndFlush(rpcResponseMessage);
            }
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            switch (event.state()) {
                case READER_IDLE:
                case WRITER_IDLE:
                case ALL_IDLE:
                    connectionPool.remove(ctx.channel());
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}
